package com.atguigu.day02.source;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import javax.sound.sampled.Port;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

public class Flink04_Source_Custom {
    public static void main(String[] args) {
         Configuration conf = new Configuration();
          //配置web页面固定端口
          conf.setInteger("rest.port", 20000);
          //创建流环境
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
          //设置并行度
          env.setParallelism(2);

          //自定义source：创建输入输出流，从socket读取数据的source
         env.addSource(new MySource("hadoop162",9999)).print();

          try {
              env.execute();
          } catch (Exception e) {
              e.printStackTrace();
          }
    }

    public static  class MySource implements SourceFunction<WaterSensor> {
        private final int port;
        private final String host;
        private Boolean cancel;
        private Socket socket;
        private BufferedReader reader;

        //socket的主机和端口由用该source的人决定，因此需要给出有参构造方法
        public MySource(String host,int port){
            this.host = host;
            this.port  = port;

        }
        //从socket读取数据的source
        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {
            System.out.println("run ...");
            //此处创建的变量需要在cancel方法中使用，因此需要将其变为成员变量，快捷键 Ctrl+Alt+f
             socket = new Socket(host, port);
            //字节流转字符流，需要一行一行的读取，再把字符流封装在字符缓冲流里
             reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String line = reader.readLine();
            while(line != null && !cancel){
                String[] data = line.split(",");//string数组
                sourceContext.collect(new WaterSensor(data[0],Long.valueOf(data[1]),Integer.valueOf(data[2])));
                line = reader.readLine();
            }
        }
        //用来关闭数据流,抛异常的方式，flink会重新运行run方法
        @Override
        public void cancel() {
            cancel = true;
            if(reader != null){
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(socket != null){
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
